package com.atguigu.gmall.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.base.BaseApp;
import com.atguigu.gmall.realtime.common.bean.TradeSkuOrderBean;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.function.BeanToJsonStrMapFunction;
import com.atguigu.gmall.realtime.common.function.DimAsyncFunction;
import com.atguigu.gmall.realtime.common.util.DateFormatUtil;
import com.atguigu.gmall.realtime.common.util.FlinkSinkUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.util.concurrent.TimeUnit;

/**
 * @author Felix
 * @date 2024/6/29
 * sku粒度下单聚合统计
 * 维度：sku
 * 度量：原始金额、优惠券减免金额、活动减免金额、实付金额
 * 需要启动的进程
 *      zk、kafka、maxwell、hdfs、hbase、redis、doris、DwdTradeOrderDetail、DwsTradeSkuOrderWindow
 * 开发流程
 *      从kafka的下单事实表中读取数据
 *      过滤空消息
 *      去重
 *          为什么会有空消息以及重复数据？
 *              下单事实表是由订单明细表、订单表、订单明细活动表、订单明细优惠券表4张表组成的
 *              这4张表在进行关联的时候，订单明细表是主表，和订单表使用内连接进行关联
 *              和订单明细活动表、订单明细优惠券表使用的左外连接进行关联的
 *              左外连接，如果左表数据先到，右表数据后到，会产生3条数据
 *                 左表   null    +I
 *                 左表   null    -D
 *                 左表   右表     +I
 *              这样的数据发送到kafka中，kafka主题会接收到3条消息
 *                  左表  null
 *                  null
 *                  左表  右表
 *             综上所述，当从下单事实表中读取数据的时候，需要过滤空消息并去重
 *          去重第一步：按照唯一键(订单明细的id)进行分组
 *          去重方式1：状态 + 定时器
 *              当第一条数据到来的时候，先不向下游传递，将其放到状态中，并注册5s后执行的定时器
 *              当第二条重复数据到来的时候，会有第二条数据的聚合时间和状态中第一条数据的聚合时间进行比较
 *              将时间大的数据放到状态中保存起来
 *              当定时器被触发的时候，将状态中的数据传递到下游
 *              优点：如果重复了，也是向下游传递一条数据，不会膨胀  缺点：时效性差
 *          去重方式2：状态 + 抵消
 *              当第一条数据到来的时候，直接传递到下游并将其放到状态中
 *              当第二条数据到来的时候，将状态中的数据(已经到下游了)影响到度量值的字段进行取反，再向下游传递
 *              将第二条数据也传递到下游
 *              优点：时效性好     缺点：如果重复了，向下游传递3条数据(数据膨胀)
 *      指定watermark以及提取事件时间字段
 *      再次对流中数据进行了转换  jsonObj->实体类对象(相当于wordcount中转换为二元组)
 *      按照统计的维度sku_id进行分组
 *      开窗
 *      聚合统计
 *      维度关联
 *          最基本的维度关联
 *              HBaseUtil->getRow
 *          优化1：旁路缓存
 *              思路：先从缓存中获取维度数据，如果获取到了，直接进行返回(缓存命中)；如果在缓存中没有找到要关联的维度
 *                  发送请求到HBase中查询维度数据，并将查询出来的维度数据放到缓存中缓存起来，方便下次查询使用
 *              缓存产品的选型：
 *                      redis    性能不错  维护性好
 *                      状态      性能更好  维护性差
 *              关于Redis的一些设置
 *                  key:    维度表名:主键值
 *                  type:   string
 *                  expire: 1day    避免冷数据常驻内存，给内存带来压力
 *                  注意：如果维度数据发生了变化 ，将redis中缓存的数据清除掉
 *                      DimSinkFunction->invoke->清除
 *              旁路缓存 + 模板
 *          优化2：异步IO
 *              为什么要使用异步IO
 *                  如果要想提升map算子的处理能力，可以调大并行度，但是更多的并行度意味着需要更多的资源，不能无限制的调整
 *                  在资源有限的情况下，可以使用异步方式关联维度
 *                  默认情况下，如果使用Map算子对流中数据进行处理，在单个并行度上，使用的是同步的处理方式
 *              发送异步请求需要具备的条件
 *                  Flink的支持
 *                      AsyncDataStream.[un]orderedWait(
 *                          流,
 *                          如何发送异步请求, implements AsyncFunction,
 *                          超时时间,
 *                          时间单位
 *                      )
 *                  Redis的支持
 *                      在RedisUtil工具类中添加了异步操作Redis的方法
 *                  HBase的支持
 *                      在HBaseUtil工具类中添加了异步操作Hbase的方法
 *              异步IO + 模板
 *                  class DimAsyncFunction extends RichAsyncFunction implements DimJoinFunction{
 *                      asynvInvoke:
 *                          //创建异步编排对象  执行线程任务 有返回值
 *                          CompletableFuture.supplyAsync-发送请求到Redis中获取维度
 *                          //执行线程任务  有入参、有返回值 - 根据结果判断是否到HBase中查询维度，并将结果放到Redis
 *                          thenApplyAsync
 *                          //执行线程任务  有入参、无返回值 ---将维度属性补充到流中对象上 并向下游传递
 *                          thenAcceptAsync
 *
 *                  }
 *      将结果写到Doris
 */
public class DwsTradeSkuOrderWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsTradeSkuOrderWindow().start(
                10029,
                4,
                "dws_trade_sku_order_window",
                Constant.TOPIC_DWD_TRADE_ORDER_DETAIL
        );
    }
    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaStrDS) {
        //TODO 1.对当前流中数据进行类型转换  并过滤空消息
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String jsonStr, ProcessFunction<String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                        if (StringUtils.isNotEmpty(jsonStr)) {
                            JSONObject jsonObj = JSON.parseObject(jsonStr);
                            out.collect(jsonObj);
                        }
                    }
                }
        );
        //jsonObjDS.print();

        //TODO 2.按照唯一键(订单明细id)进行分组
        KeyedStream<JSONObject, String> orderDetailIdKeyedDS = jsonObjDS.keyBy(jsonObj -> jsonObj.getString("id"));
        //TODO 3.去重
        /*//3.1 去重方式1：状态 + 定时器   缺点：不管数据是否重复 ，都需要等待5s后才能向下游传递，时效性差   优点：如果有重复数据产生，只会向下游发送一条数据
        SingleOutputStreamOperator<JSONObject> distinctDS = orderDetailIdKeyedDS.process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                    private ValueState<JSONObject> lastJsonObjState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<JSONObject> valueStateDescriptor
                                = new ValueStateDescriptor<JSONObject>("lastJsonObjState", JSONObject.class);
                        lastJsonObjState = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, KeyedProcessFunction<String, JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                        //从状态中获取数据
                        JSONObject lastJsonObj = lastJsonObjState.value();
                        if (lastJsonObj == null) {
                            //没重复
                            //将当前数据放到状态中
                            lastJsonObjState.update(jsonObj);
                            //注册一个5s后执行的定时器
                            long currentProcessingTime = ctx.timerService().currentProcessingTime();
                            ctx.timerService().registerProcessingTimeTimer(currentProcessingTime + 5000L);
                        } else {
                            //重复 ----伪代码
                            //获取状态中已经到来的数据的关联时间
                            String ts1 = lastJsonObj.getString("关联的时间");
                            //获取当前数据的关联时间
                            String ts2 = jsonObj.getString("关联的时间");
                            if (ts2.compareTo(ts1) >= 0) {
                                //将当前数据放到状态中
                                lastJsonObjState.update(jsonObj);
                            }
                        }
                    }

                    @Override
                    public void onTimer(long timestamp, KeyedProcessFunction<String, JSONObject, JSONObject>.OnTimerContext ctx, Collector<JSONObject> out) throws Exception {
                        //定时器被触发的时候，执行的方式
                        JSONObject lastJsonObj = lastJsonObjState.value();
                        out.collect(lastJsonObj);
                        lastJsonObjState.clear();
                    }
                }
        );*/
        //3.2 去重方式2：状态 + 抵消         优点：时效性好     缺点：如果有重复数据产生，会向下游发送3条数据(数据膨胀)
        SingleOutputStreamOperator<JSONObject> distinctDS = orderDetailIdKeyedDS.process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                    private ValueState<JSONObject> lastJsonObjState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<JSONObject> valueStateDescriptor
                                = new ValueStateDescriptor<JSONObject>("lastJsonObjState", JSONObject.class);
                        valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.seconds(10)).build());
                        lastJsonObjState = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, KeyedProcessFunction<String, JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                        //从状态中获取数据
                        JSONObject lastJsonObj = lastJsonObjState.value();

                        if (lastJsonObj != null) {
                            //重复了 将状态中(已经发送到下游了) 影响到度量值的字段 进行取反，再向下游发送
                            String splitOriginalAmount = lastJsonObj.getString("split_original_amount");
                            String splitCouponAmount = lastJsonObj.getString("split_coupon_amount");
                            String splitActivityAmount = lastJsonObj.getString("split_activity_amount");
                            String splitTotalAmount = lastJsonObj.getString("split_total_amount");

                            lastJsonObj.put("split_original_amount", "-" + splitOriginalAmount);
                            lastJsonObj.put("split_coupon_amount", "-" + splitCouponAmount);
                            lastJsonObj.put("split_activity_amount", "-" + splitActivityAmount);
                            lastJsonObj.put("split_total_amount", "-" + splitTotalAmount);
                            out.collect(lastJsonObj);
                        }
                        out.collect(jsonObj);
                        lastJsonObjState.update(jsonObj);
                    }
                }
        );

        //distinctDS.print();

        //TODO 4.指定Watermark以及提取事件时间字段
        SingleOutputStreamOperator<JSONObject> withWatermarkDS = distinctDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<JSONObject>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<JSONObject>() {
                                    @Override
                                    public long extractTimestamp(JSONObject jsonObj, long recordTimestamp) {
                                        return jsonObj.getLong("ts") * 1000;
                                    }
                                }
                        )
        );
        //TODO 5.再次对流中数据进行类型转换  jsonObj->实体类对象
        SingleOutputStreamOperator<TradeSkuOrderBean> beanDS = withWatermarkDS.map(
                new MapFunction<JSONObject, TradeSkuOrderBean>() {
                    @Override
                    public TradeSkuOrderBean map(JSONObject jsonObj) throws Exception {
                        //{"create_time":"2024-06-28 11:19:27","sku_num":"1","activity_rule_id":"1","split_original_amount":"6999.0000",
                        // "split_coupon_amount":"0.0","sku_id":"2","date_id":"2024-06-28","user_id":"2447","province_id":"30",
                        // "activity_id":"1","sku_name":"小米12S","id":"13020","order_id":"9243","split_activity_amount":"500.0",
                        // "split_total_amount":"6499.0","ts":1719631167}
                        String skuId = jsonObj.getString("sku_id");
                        BigDecimal splitOriginalAmount = jsonObj.getBigDecimal("split_original_amount");
                        BigDecimal splitCouponAmount = jsonObj.getBigDecimal("split_coupon_amount");
                        BigDecimal splitActivityAmount = jsonObj.getBigDecimal("split_activity_amount");
                        BigDecimal splitTotalAmount = jsonObj.getBigDecimal("split_total_amount");
                        Long ts = jsonObj.getLong("ts") * 1000;
                        TradeSkuOrderBean orderBean = TradeSkuOrderBean.builder()
                                .skuId(skuId)
                                .originalAmount(splitOriginalAmount)
                                .couponReduceAmount(splitCouponAmount)
                                .activityReduceAmount(splitActivityAmount)
                                .orderAmount(splitTotalAmount)
                                .ts(ts)
                                .build();
                        return orderBean;
                    }
                }
        );

        //beanDS.print();

        //TODO 6.分组---按照统计的维度sku进行分组
        KeyedStream<TradeSkuOrderBean, String> skuIdKeyedDS = beanDS.keyBy(TradeSkuOrderBean::getSkuId);

        //TODO 7.开窗
        WindowedStream<TradeSkuOrderBean, String, TimeWindow> windowDS
                = skuIdKeyedDS.window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)));

        //TODO 8.聚合
        SingleOutputStreamOperator<TradeSkuOrderBean> reduceDS = windowDS.reduce(
                new ReduceFunction<TradeSkuOrderBean>() {
                    @Override
                    public TradeSkuOrderBean reduce(TradeSkuOrderBean value1, TradeSkuOrderBean value2) throws Exception {
                        value1.setOriginalAmount(value1.getOriginalAmount().add(value2.getOriginalAmount()));
                        value1.setCouponReduceAmount(value1.getCouponReduceAmount().add(value2.getCouponReduceAmount()));
                        value1.setActivityReduceAmount(value1.getActivityReduceAmount().add(value2.getActivityReduceAmount()));
                        value1.setOrderAmount(value1.getOrderAmount().add(value2.getOrderAmount()));
                        return value1;
                    }
                },
                new WindowFunction<TradeSkuOrderBean, TradeSkuOrderBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<TradeSkuOrderBean> input, Collector<TradeSkuOrderBean> out) throws Exception {
                        TradeSkuOrderBean orderBean = input.iterator().next();
                        String stt = DateFormatUtil.tsToDateTime(window.getStart());
                        String edt = DateFormatUtil.tsToDateTime(window.getEnd());
                        String curDate = DateFormatUtil.tsToDate(window.getStart());

                        orderBean.setStt(stt);
                        orderBean.setEdt(edt);
                        orderBean.setCurDate(curDate);
                        out.collect(orderBean);
                    }
                }

        );
        //reduceDS.print();
        //TODO 9.关联sku维度
        /*
        //维度关联的最基本的实现方式
        SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = reduceDS.map(
                new RichMapFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {
                    private Connection hbaseConn;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        hbaseConn = HBaseUtil.getHBaseConnection();
                    }

                    @Override
                    public void close() throws Exception {
                        HBaseUtil.closeHBaseConnection(hbaseConn);
                    }

                    @Override
                    public TradeSkuOrderBean map(TradeSkuOrderBean orderBean) throws Exception {
                        //根据流中的对象获取要关联的维度的主键
                        String skuId = orderBean.getSkuId();
                        //根据维度的主键获取对应的维度对象
                        JSONObject dimJsonObj = HBaseUtil.getRow(hbaseConn, Constant.HBASE_NAMESPACE, "dim_sku_info", skuId, JSONObject.class);
                        //将维度对象相关的维度属性补充到流中的对象上
                        //id,spu_id,price,sku_name,sku_desc,weight,tm_id,category3_id,sku_default_img,is_sale,create_time
                        orderBean.setSkuName(dimJsonObj.getString("sku_name"));
                        orderBean.setSpuId(dimJsonObj.getString("spu_id"));
                        orderBean.setTrademarkId(dimJsonObj.getString("tm_id"));
                        orderBean.setCategory3Id(dimJsonObj.getString("category3_id"));
                        return orderBean;
                    }
                }
        );
        withSkuInfoDS.print();

        //优化1：旁路缓存
        SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = reduceDS.map(
                new RichMapFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {

                    private Connection hbaseConn;
                    private Jedis jedis;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        hbaseConn = HBaseUtil.getHBaseConnection();
                        jedis = RedisUtil.getJedis();
                    }

                    @Override
                    public void close() throws Exception {
                        HBaseUtil.closeHBaseConnection(hbaseConn);
                        RedisUtil.closeJedis(jedis);
                    }


                    @Override
                    public TradeSkuOrderBean map(TradeSkuOrderBean orderBean) throws Exception {
                        //根据流中对象获取要关联的维度的主键
                        String skuId = orderBean.getSkuId();
                        //先从redis中获取对应的维度数据
                        JSONObject dimJsonObj = RedisUtil.readDim(jedis, "dim_sku_info", skuId);
                        if (dimJsonObj != null) {
                            //如果获取到了，直接将其作为维度进行返回(缓存命中)
                            System.out.println("~~~从Redis中获取到了维度数据~~~");
                        } else {
                            //如果从redis中没有获取到对应的维度数据，发送请求从HBase中查询维度
                            dimJsonObj = HBaseUtil.getRow(hbaseConn, Constant.HBASE_NAMESPACE, "dim_sku_info", skuId, JSONObject.class);
                            if (dimJsonObj != null) {
                                //将查询到的维度数据放到Redis中缓存起来，方便下次查询使用
                                System.out.println("~~~从HBase中获取到了维度数据~~~");
                                RedisUtil.writeDim(jedis, "dim_sku_info", skuId, dimJsonObj);
                            } else {
                                System.out.println("~~~没有找到要关联的维度~~~");
                            }
                        }

                        if (dimJsonObj != null) {
                            //将维度属性补充到流中的对象上
                            orderBean.setSkuName(dimJsonObj.getString("sku_name"));
                            orderBean.setSpuId(dimJsonObj.getString("spu_id"));
                            orderBean.setTrademarkId(dimJsonObj.getString("tm_id"));
                            orderBean.setCategory3Id(dimJsonObj.getString("category3_id"));
                        }
                        return orderBean;
                    }
                }
        );
        //withSkuInfoDS.print();

        //旁路缓存 + 模板方法
        SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = reduceDS.map(
                new DimMapFunction<TradeSkuOrderBean>() {
                    @Override
                    public void addDims(TradeSkuOrderBean orderBean, JSONObject dimJsonObj) {
                        orderBean.setSkuName(dimJsonObj.getString("sku_name"));
                        orderBean.setSpuId(dimJsonObj.getString("spu_id"));
                        orderBean.setTrademarkId(dimJsonObj.getString("tm_id"));
                        orderBean.setCategory3Id(dimJsonObj.getString("category3_id"));
                    }

                    @Override
                    public String getTableName() {
                        return "dim_sku_info";
                    }

                    @Override
                    public String getRowKey(TradeSkuOrderBean orderBean) {
                        return orderBean.getSkuId();
                    }
                }
        );

        withSkuInfoDS.print();

        //优化2：异步IO
        //将异步I/O操作应用于DataStream作为DataStream的一次转换操作
        SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = AsyncDataStream.unorderedWait(
                reduceDS,
                //如何发送异步请求，实现分发请求的AsyncFunction
                new RichAsyncFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {
                    private StatefulRedisConnection<String,String> redisAsyncConn;
                    private AsyncConnection hbaseAsyncConn;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        redisAsyncConn = RedisUtil.getAsyncRedisConnection();
                        hbaseAsyncConn = HBaseUtil.getAsyncHBaseConnection();
                    }

                    @Override
                    public void close() throws Exception {
                        RedisUtil.closeAsyncRedisConnection(redisAsyncConn);
                        HBaseUtil.closeAsyncHBaseConnection(hbaseAsyncConn);
                    }

                    @Override
                    public void asyncInvoke(TradeSkuOrderBean orderBean, ResultFuture<TradeSkuOrderBean> resultFuture) throws Exception {
                        //根据流中对象获取要关联的维度的主键
                        String skuId = orderBean.getSkuId();
                        //发送异步请求到Redis中获取维度数据
                        JSONObject dimJsonObj = RedisUtil.readDimAsync(redisAsyncConn, "dim_sku_info", skuId);
                        if(dimJsonObj != null){
                            //如果在Redis中找到了要关联的维度，直接将其返回(缓存命中)
                            System.out.println("~~~从Redis中获取维度~~~");
                        }else{
                            //如果在Redis中没有找到要关联的维度，发送异步请求到HBase中查询维度数据
                            dimJsonObj = HBaseUtil.readDimAsync(hbaseAsyncConn,Constant.HBASE_NAMESPACE,"dim_sku_info", skuId);
                            if(dimJsonObj != null){
                                //将查询出来的维度以异步的方式放到Redis中保存起来，方便下次查询使用
                                System.out.println("~~~从HBase中获取维度~~~");
                                RedisUtil.writeDimAsync(redisAsyncConn,"dim_sku_info",skuId,dimJsonObj);
                            }else{
                                System.out.println("~~~没有找到对应的维度~~~");
                            }
                        }
                        //将维度属性补充到流中对象上
                        if(dimJsonObj != null){
                            orderBean.setSkuName(dimJsonObj.getString("sku_name"));
                            orderBean.setSpuId(dimJsonObj.getString("spu_id"));
                            orderBean.setTrademarkId(dimJsonObj.getString("tm_id"));
                            orderBean.setCategory3Id(dimJsonObj.getString("category3_id"));
                        }
                        //获取数据库交互的结果并发送给ResultFuture的回调函数  向下游传递数据
                        resultFuture.complete(Collections.singleton(orderBean));
                    }
                },
                60,
                TimeUnit.SECONDS
        );
        //withSkuInfoDS.print();
         */
        //旁路缓存 + 异步IO + 模板方法设计模式
        SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = AsyncDataStream.unorderedWait(
                reduceDS,
                new DimAsyncFunction<TradeSkuOrderBean>() {
                    @Override
                    public void addDims(TradeSkuOrderBean orderBean, JSONObject dimJsonObj) {
                        orderBean.setSkuName(dimJsonObj.getString("sku_name"));
                        orderBean.setSpuId(dimJsonObj.getString("spu_id"));
                        orderBean.setTrademarkId(dimJsonObj.getString("tm_id"));
                        orderBean.setCategory3Id(dimJsonObj.getString("category3_id"));
                    }

                    @Override
                    public String getTableName() {
                        return "dim_sku_info";
                    }

                    @Override
                    public String getRowKey(TradeSkuOrderBean orderBean) {
                        return orderBean.getSkuId();
                    }
                },
                60,
                TimeUnit.SECONDS
        );
        //withSkuInfoDS.print();
        //TODO 10.关联spu维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withSpuInfoDS = AsyncDataStream.unorderedWait(
                withSkuInfoDS,
                new DimAsyncFunction<TradeSkuOrderBean>() {
                    @Override
                    public void addDims(TradeSkuOrderBean orderBean, JSONObject dimJsonObj) {
                        orderBean.setSpuName(dimJsonObj.getString("spu_name"));
                    }

                    @Override
                    public String getTableName() {
                        return "dim_spu_info";
                    }

                    @Override
                    public String getRowKey(TradeSkuOrderBean orderBean) {
                        return orderBean.getSpuId();
                    }
                },
                60,
                TimeUnit.SECONDS
        );

        //TODO 11.关联tm维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withTmDS = AsyncDataStream.unorderedWait(
                withSpuInfoDS,
                new DimAsyncFunction<TradeSkuOrderBean>() {
                    @Override
                    public void addDims(TradeSkuOrderBean orderBean, JSONObject dimJsonObj) {
                        orderBean.setTrademarkName(dimJsonObj.getString("tm_name"));
                    }

                    @Override
                    public String getTableName() {
                        return "dim_base_trademark";
                    }

                    @Override
                    public String getRowKey(TradeSkuOrderBean orderBean) {
                        return orderBean.getTrademarkId();
                    }
                },
                60,
                TimeUnit.SECONDS
        );
        //TODO 12.关联category3维度
        SingleOutputStreamOperator<TradeSkuOrderBean> c3Stream = AsyncDataStream.unorderedWait(
                withTmDS,
                new DimAsyncFunction<TradeSkuOrderBean>() {
                    @Override
                    public String getRowKey(TradeSkuOrderBean bean) {
                        return bean.getCategory3Id();
                    }

                    @Override
                    public String getTableName() {
                        return "dim_base_category3";
                    }

                    @Override
                    public void addDims(TradeSkuOrderBean bean, JSONObject dim) {
                        bean.setCategory3Name(dim.getString("name"));
                        bean.setCategory2Id(dim.getString("category2_id"));
                    }
                },
                120,
                TimeUnit.SECONDS
        );

        //TODO 13.关联category2维度
        SingleOutputStreamOperator<TradeSkuOrderBean> c2Stream = AsyncDataStream.unorderedWait(
                c3Stream,
                new DimAsyncFunction<TradeSkuOrderBean>() {
                    @Override
                    public String getRowKey(TradeSkuOrderBean bean) {
                        return bean.getCategory2Id();
                    }

                    @Override
                    public String getTableName() {
                        return "dim_base_category2";
                    }

                    @Override
                    public void addDims(TradeSkuOrderBean bean, JSONObject dim) {
                        bean.setCategory2Name(dim.getString("name"));
                        bean.setCategory1Id(dim.getString("category1_id"));
                    }
                },
                120,
                TimeUnit.SECONDS
        );

        //TODO 14.关联category1维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withC1DS = AsyncDataStream.unorderedWait(
                c2Stream,
                new DimAsyncFunction<TradeSkuOrderBean>() {
                    @Override
                    public String getRowKey(TradeSkuOrderBean bean) {
                        return bean.getCategory1Id();
                    }

                    @Override
                    public String getTableName() {
                        return "dim_base_category1";
                    }

                    @Override
                    public void addDims(TradeSkuOrderBean bean, JSONObject dim) {
                        bean.setCategory1Name(dim.getString("name"));
                    }
                },
                120,
                TimeUnit.SECONDS
        );
        withC1DS.print();

        //TODO 15.将关联的结果写到Doris中
        withC1DS
                .map(new BeanToJsonStrMapFunction<>())
                .sinkTo(FlinkSinkUtil.getDorisSink("dws_trade_sku_order_window"));

    }
}
